package cn.smileyan.demo;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.utils.MultipleParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 *
 * @author Smileyan
 */
public class GenericFlinkElasticsearchDemo {
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final MultipleParameterTool parameters = MultipleParameterTool.fromArgs(args);

        // 设置 Elasticsearch 集群的连接信息 比如 http://localhost:9200 | http://localhost:9200,http://localhost:9200
        String urls = parameters.get("es.hosts", "http://localhost:9200");
        final String regex = ",";
        List<HttpHost> httpHosts = Arrays.stream(urls.split(regex)).map(HttpHost::create).collect(Collectors.toList());

        // 创建 ElasticsearchSinkFunction 用于将数据写入 Elasticsearch
        final String index = parameters.get("es.index", "hello-world-2");

        GenericElasticsearchSinkFunction<Student> elasticsearchSinkFunction = new GenericElasticsearchSinkFunction<>(index);
        // 创建 ElasticsearchSink
        ElasticsearchSink.Builder<Student> esSinkBuilder = new ElasticsearchSink.Builder<>(httpHosts, elasticsearchSinkFunction);

        DataStream<Student> inputDataStream = env.fromElements(
                new Student(1, "张三", 18),
                new Student(2, "李四", 20),
                new Student(3, "王五", 22)
        );

        // 将结果 Sink 到 Elasticsearch
        inputDataStream.addSink(esSinkBuilder.build());

        // 执行 Flink 作业
        env.execute("Flink Elasticsearch Example");
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }
}
